package action.RDD基本转换操作2

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Coalesce操作
  * coalesce(numPartitions)
  * Decrease the number of partitions in the RDD to numPartitions.
  * Useful for running operations more efficiently after filtering down a large dataset
  *
  * 将RDD中的分区数量<b>减少</b>到numpartition
  *
  * @author wdmcode@aliyun.com
  * @version 1.0.0
  * @date 2018/11/8
  */
object SparkCoalesce {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("SparkParallelize")
    conf.setMaster("local[2]")

    val spark = new SparkContext(conf)

    val rdd = spark.parallelize(1 to 10)
    println(rdd.partitions.size) // 2
    /**
      * 方法注释
      * 返回一个经过简化到numPartitions个分区的新RDD。
      * 这会导致一个窄依赖，例如：你将1000个分区转换成100个分区，这个过程不会发生shuffle，
      * 相反如果10个分区转换成100个分区将会发生shuffle。
      * 然而如果你想大幅度合并分区，例如合并成一个分区，
      * 这会导致你的计算在少数几个集群节点上计算（言外之意：并行度不够）。
      * 为了避免这种情况，你可以将第二个shuffle参数传递一个true，
      * 这样会在重新分区过程中多一步shuffle，这意味着上游的分区可以并行运行。
      */
    /**
      * 注意：
      * shuffle为false coalesce只支持减少分区数，不支持增加分区数
      *
      * 注意：第二个参数shuffle=true，将会产生多于之前的分区数目，
      * 例如你有一个个数较少的分区，假如是100，调用coalesce(1000, shuffle = true)
      * 将会使用一个  HashPartitioner产生1000个分区分布在集群节点上。
      * 这个（对于提高并行度）是非常有用的。
      *
      */
    val coalRDD = rdd.coalesce(1)

    println(coalRDD.partitions.size)

    /**
      * 当shuffle为TRUE时
      */
    val calRddAndShuffle = rdd.coalesce(3, true)

    println(calRddAndShuffle.partitions.size)

    spark.stop()
  }


}
